package com.kafka.xiang.kakfamodual;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.common.Node;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Collection;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class KafkaConsumerOperation {

    private Logger logger = LoggerFactory.getLogger(KafkaConsumerOperation.class);

    private volatile boolean active = false;
    private MessageHandler messageHandler;
    private boolean consumerCreationSuccessful=false;
    public void terminateConsumer() {
        active = false;
    }

    public void createConsumer(String topics, String bootstrapServers, String groupId, boolean consumerActive) {
        System.out.println(topics+bootstrapServers+groupId+consumerActive);
        active = consumerActive;//是否一直持续监听

        Properties props = new Properties();//Java 中的 Properties 类用于存储键值对，用于配置 Kafka 消费者的属性。
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); // 替换为实际的值的反序列化器类

        // 创建 AdminClient， 用于执行集群管理任务
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 调用 describeCluster 方法
            DescribeClusterResult describeClusterResult = adminClient.describeCluster();

            try {
                // 获取 describeCluster 的结果
                Collection<Node> nodes = describeClusterResult.nodes().get();//可能获取不到结果，表明没有创建成功

                // 打印集群信息
                System.out.println("Connected to Kafka successfully!");
                System.out.println("Brokers: ");
                for (Node node : nodes) {
                    System.out.println("\t" + node);
                }
                this.consumerCreationSuccessful=true;
                //连接成功，创建消费者。
                try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
                    consumer.subscribe(Collections.singletonList(topics));
                    //Set<String> subscribedTopics = consumer.subscription();//consumer.subscription()返回订阅的主题。
                    //System.out.println(subscribedTopics.toString());
                    //if(subscribedTopics.contains(topics)) this.consumerCreationSuccessful=true;//成功订阅
                    while (active) {
                        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1));//使用 Duration 来设置 poll 方法的超时，以毫秒为单位

                        records.forEach(record -> {//在这个 Lambda 表达式中，record 是参数（每个元素都被赋给这个参数），而 {} 中的部分是执行的操作。
                            try {

                                String key = record.key();
                                String value = record.value();

                                logger.info("Received message: {} - {}", key, value);

                                if (messageHandler != null) {
                                    messageHandler.handleMessage(key, value);
                                }
                            }catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
                    }
                    // 关闭消费者
                    consumer.close();
                }
            } catch (InterruptedException | ExecutionException e) {
                System.out.println("尚未成功连接诶");
                adminClient.close();
                e.printStackTrace();//没有成功连接，无法输出集群消息。
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public void setMessageHandler(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }

    public interface MessageHandler {
        void handleMessage(String key, String value);
    }

    // Add other methods and configurations as needed
    public boolean getconsumerCreationSuccessful(){
        return this.consumerCreationSuccessful;
    }
    public boolean setActive(boolean active) {
        this.active = active;
        return this.active;
    }
}
